package com.asiainfo.dacp.datastash.broker.workerctrl;

import java.io.File;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.asiainfo.dacp.datastash.broker.core.BrokerException;
import com.asiainfo.dacp.datastash.broker.core.BrokerModel;
import com.asiainfo.dacp.datastash.broker.core.enums.UnitStepTypeEnum;
import com.asiainfo.dacp.datastash.broker.core.vo.StashUnitTask;
import com.asiainfo.dacp.datastash.broker.facade.UnitProcessFacade;
import com.asiainfo.dacp.datastash.broker.facade.UnitTaskFacade;
import com.asiainfo.dacp.datastash.broker.util.CallableResult;
import com.asiainfo.dacp.datastash.broker.util.ClassLoaderSwapper;
import com.asiainfo.dacp.datastash.broker.util.Configuration;
import com.asiainfo.dacp.datastash.broker.util.JarLoader;
import com.asiainfo.dacp.datastash.broker.util.ThreadEntity;
import com.asiainfo.dacp.datastash.broker.workerctrl.service.ResultCheckService;
import com.asiainfo.dacp.datastash.common.WorkerExecutor;
import com.asiainfo.dacp.datastash.worker.executor.adapter.datax.DataxWorkerExecutor;
import com.asiainfo.dacp.datastash.worker.executor.adapter.datax.DpWorkerExecutor;
import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
import com.google.common.collect.Maps;

@Service
public class StashWorkExecutorByEngine {
	private static Logger LOG = LoggerFactory.getLogger(StashWorkExecutorByEngine.class);
	/**
	 * 插件统一按引擎放置 engine/dp engine/datax engine/realtime，插件按需安装 各自配置Json
	 * 文件，配置引擎入口类
	 * 
	 */
	public static String DATASTASH_HOME = System.getProperty("datastash.home");
	public static String DATAX_CONF_PATH = DATASTASH_HOME + File.separator + "engine";

	private static Map<String, JarLoader> jarLoaderCenter = new HashMap<String, JarLoader>();

	public static Map<String, Configuration> pluginRegisterCenter = Maps.newConcurrentMap();


	private ExecutorService cachedThreadPool = Executors.newCachedThreadPool();


	@Autowired
	private UnitTaskFacade unitTaskFacade;
	@Autowired
	private UnitProcessFacade unitProcessFacade;
	@Autowired
	private BrokerModel brokerModel;
	@Autowired
	private ResultCheckService resultCheck;

	private void init() {

		Preconditions.checkArgument(!Strings.isNullOrEmpty(DATASTASH_HOME),
				String.format("Please specify the parameters -Ddatastash.home ."));
		String[] engines = new File(DATAX_CONF_PATH).list();
		Preconditions.checkNotNull(engines);
		for (final String each : engines) {
			try {
				loadEngine(each);
			} catch (Exception e) {
				LOG.error(String.format("Load engine [%s] failure .", each), e);
				continue;
			}
		}
	}

	private static void loadEngine(String type) {
		String basePath = DATAX_CONF_PATH + File.separator + type;
		String engineCfgFile = basePath + File.separator + "engine.json";
		String pluginPath = basePath + File.separator + "lib";
		try {
			pluginRegisterCenter.put(type, Configuration.from(new File(engineCfgFile)));
		} catch (BrokerException e) {
			e.printStackTrace();
		}
		jarLoaderCenter.put(type, new JarLoader(new String[] { pluginPath }));
	}

	String type(StashUnitTask stashUnitShard) {
		// TODO 根据任务判断引擎类型
		return "datax";
	}

	@SuppressWarnings("unchecked")
	public void innerExecute(StashUnitTask stashUnitShard) throws Throwable {
		if (jarLoaderCenter.isEmpty()) {
			init();
		}
		String type = type(stashUnitShard);
		final JarLoader jarLoader = jarLoaderCenter.get(type);
		if (jarLoader == null) {
			// 尝试重新加载 热部署
			loadEngine(type);
		}
		Preconditions.checkNotNull(jarLoader, "No appropriate execution engine was found .");
		ClassLoaderSwapper classLoaderSwapper = ClassLoaderSwapper.newCurrentThreadClassLoaderSwapper();
		classLoaderSwapper.setCurrentThreadClassLoader(jarLoader);  
		
		Class<? extends WorkerExecutor> clazz = (Class<? extends WorkerExecutor>) jarLoader
				.loadClass(pluginRegisterCenter.get(type).getString("class"));
		Preconditions.checkNotNull(clazz, "Load execution engine class failure .");
		WorkerExecutor executor = (WorkerExecutor) clazz.newInstance();
		executor.execute(new String[] { JSON.toJSONString(stashUnitShard, SerializerFeature.PrettyFormat) });

		
		 
		classLoaderSwapper.restoreCurrentThreadClassLoader();
	}

	public void execute(final StashUnitTask stashUnitTask) {

		//unitTaskFacade.taskProcess(stashUnitTask);


		WorkerExecutor executor=null;
		if(stashUnitTask.getTaskType().equalsIgnoreCase(UnitStepTypeEnum.DP.value())) {
			executor=new DpWorkerExecutor();
			//executor=new MockDpWorkerExecutor();
		}else if(stashUnitTask.getTaskType().equalsIgnoreCase(UnitStepTypeEnum.Datax.value())){
			executor=new DataxWorkerExecutor();
			//executor=new MockDataxWorkerExecutor();
		}
		Future<CallableResult> future=cachedThreadPool.submit(new ThreadEntity(executor,new String[] { JSON.toJSONString(stashUnitTask, SerializerFeature.PrettyFormat) }));
		
		
		resultCheck.add(stashUnitTask, future);
		
		/*
		CallableResult result=null;
		try {
			result = future.get();
			if(result.isError()){
				//TODO  后续考虑失败重试 ，需表结构调整  task_state 加上 retry_cnt 字段
				unitTaskFacade.taskFailed(stashUnitTask);
				//失败
			}else{
				unitTaskFacade.taskComplete(stashUnitTask);
			}
			unitProcessFacade.checkUnitResult(stashUnitTask);
			brokerModel.deleteBrokerUnitTaskList(stashUnitTask.getTaskProcId());
			
		}catch (Exception e){
			LOG.error(e.getMessage());
			e.printStackTrace();
		}
		*/
	}
}
